package xj.toolkit.netty.transport.endpoint;

import org.apache.commons.pool2.ObjectPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import xj.toolkit.base.Sender;
import xj.toolkit.netty.transport.PoolableSender;

import javax.annotation.PostConstruct;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 使用时可以按需创建连接并存放在连接池中的Endpoint
 * @param <I>
 */
public class PoolableEndpoint<I> implements Sender<I> {
    private static Logger logger = LoggerFactory.getLogger(PoolableEndpoint.class);

    ObjectPool<PoolableSender<I>> pool;
    private long waitTimeout = 1000;
    private BlockingQueue<I> sendQueue = null;

    private ExecutorService exec = Executors.newSingleThreadExecutor();

    @PostConstruct
    public void init() {
        doSendPending();
    }

    public void send(I bean) {
        if (null != this.sendQueue) {
            if (logger.isTraceEnabled()) {
                logger.trace("send bean:" + bean);
            }
            sendQueue.add(bean);
        } else {
            logger.warn("Unsupport Operation for send.");
        }
    }

    public void stop() {
        if (null != this.sendQueue) {
            this.sendQueue.clear();
        }
    }

    private void doSendPending() {
        exec.submit(new Runnable() {

            public void run() {
                sendPending();
            }
        });
    }

    private void sendPending() {
        try {
            I bean = sendQueue.poll(waitTimeout, TimeUnit.MILLISECONDS);
            if (null != bean) {
                PoolableSender<I> client = null;
                for (int i = 0; i < 5; i++) {
                    client = pool.borrowObject();
                    client.setReturnedToPool(false);
                    logger.debug("borrowed {} from pool", client);
                    if (client.getChannel() == null || !client.getChannel().isActive() && !client.isReturnedToPool()) {
                        logger.debug("client {} is not connected,drop it to pool", client);
                        client.setReturnedToPool(true);
                        pool.returnObject(client);
                    } else {
                        break;
                    }
                }
                if (client != null) {
                    logger.debug("use client {} to send msg {}", client, bean);
                    client.send(bean);
                } else {
                    logger.error("can't obtain available client from pool");
                }
            }
        } catch (Exception e) {
            logger.error("sendPending error:", e);
        } finally {
            doSendPending();
        }
    }

    public int getPendingCount() {
        if (null != this.sendQueue) {
            return this.sendQueue.size();
        } else {
            return -1;
        }
    }

    public void setWaitTimeout(long waitTimeout) {
        this.waitTimeout = waitTimeout;
    }

    public void setSendQueue(BlockingQueue<I> sendQueue) {
        this.sendQueue = sendQueue;
    }

    public void setPool(ObjectPool<PoolableSender<I>> pool) {
        this.pool = pool;
    }

	@Override
	public void send(Object req, I response) {
		this.send(response);
	}
}
